package org.example.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * rdd的算子，也就是rdd的方法
 *
 *
 */
object Spark15_RDD_Operator_GroupByKey {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val context = new SparkContext(conf)
    // key value 类型
    val rdd1 = context.makeRDD(List(
      ("a",1),("a",2),("a",3),("b",4)),2)

    /**
      * 将数据源中的数据，相同key的数据分成一个组，
      * 形成一个对偶元祖
      * 元祖的第一个元素就是key
      * 元祖中的第二个元素就是相同key的value集合
      *
      *
      */
    val groupRdd: RDD[(String, Iterable[Int])] = rdd1.groupByKey()

    /**
      * 之前我们学过一种叫做groupby的方式
      */
    val grRdd: RDD[(String, Iterable[(String, Int)])] = rdd1.groupBy(_._1)
    // groupByKey是一定通过key来分组，但是groupBy是不确定的
    // 所以他们最后的分组结果是元素本身
    /**
      * 我们可以再以这两个做例子，reducebykey和groupbykey
      *
      *
      *
      * 两者都存在shuffle操作，即打乱分区内的数据，
      * 但是这回引发一个问题，那就是如果一号分区
      * (a,1) (a,2) (a,3)  (b,1)
      * 二号分区
      * (b,2)(b,3)(b,4) (a,4)
      *
      * 那么在按照key来分组的时候，一号分区里存在的b，1是二号分区所需要的数据
      * 也就是说原本两个没有关联的分区，在这里产生了关联，也就会说，二号分区必须要等一号分区处理完毕后，让b，1
      * 落到二号分区时，才可以进行计算。那么这里会存在等待问题，具体等待多少是未知的
      * 由于数据会持续堆积在内存中，会造成溢出。
      * 所以在进行下一个操作之前，会先进行落盘操作，写入磁盘文件
      * 等待所有写入完毕后，会再次写回内存进行下一个的操作。
      * 由于有磁盘的交互，所以会存在效率问题。
      *
      * 所以groupbykey会对还未落盘的数据，进行一次聚合来降低
      * 写入的数据量，由于数据量小了，读写速度会提升。
      * 这一步叫做combine，预聚合操作
      *
      * 从功能上来讲，
      * reducebykey可以用于分组和聚合
      * groupbykey只可以用来做分组，所以如果你只想要分区，不需要
      * 聚合，那么groupbykey是最好的选择
      */
    context.stop()
  }
}
